package cn.xuexiyuan.flinkstudy.sink;

import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Description: 基于控制台和文件的 Sink
 * @Author 左龙龙
 * @Date 21-3-24
 * @Version 1.0
 **/
public class SinkDemo01 {

    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<String> ds = env.readTextFile("data/input/word1.txt");

        // 2.transformation

        // 3.sink
        // 直接输出到控制台
        ds.print("INFO");
        // 会在控制台以红色文字输出
        ds.printToErr("ERROR");
        // 输出到文本文件,目录不存在时会自动创建
        // 并行度 > 1, 则path 为目录， 并行度=1, 则path为文件名
        ds.writeAsText("data/output/result1.txt", FileSystem.WriteMode.OVERWRITE).setParallelism(1);
        ds.writeAsText("data/output/result2", FileSystem.WriteMode.OVERWRITE).setParallelism(2);

        // 4.excute
        env.execute("SinkDemo01");

    }

}
